package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

var NReduce int

type worker struct {
	id      int
	mapf    func(string, string) []KeyValue
	reducef func(string, []string) string
}

func (w *worker) Run() {
	for {
		isDone := w.doMap()
		if isDone == NEXT {
			break
		} else if isDone == HOLDON || isDone == ERROR {
			time.Sleep(time.Second)
		}
	}
	for {
		isDone := w.doReduce()
		if isDone == NEXT {
			break
		} else if isDone == HOLDON || isDone == ERROR {
			time.Sleep(time.Second)
		}
	}
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	work := &worker{
		id:      1,
		mapf:    mapf,
		reducef: reducef,
	}
	work.Run()
}

func (w *worker) doMap() int {
	// 获取 RPC
	args := MapArgs{RequestType: 0}
	reply := MapReply{}
	call("Master.GetMap", &args, &reply)
	if reply.Status != SUCCESS {
		return reply.Status
	}
	// uncomment to send the Example RPC to the master.
	filename := reply.Filename
	workid := reply.Workid
	NReduce = reply.NReduce

	//go func() {
	//	args := HeartBeatArgs{}
	//	reply := HeartBeatReply{}
	//	call("Master.HeartBeat", &args, &reply)
	//}()

	// 打开指定的文件
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("in map cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", filename)
	}
	_ = file.Close()

	// map阶段开始

	// 调用 mapf函数，并排序
	kva := w.mapf(filename, string(content))
	sort.Sort(ByKey(kva))

	// 写入中间文件
	cacheFiles := make(map[string]*json.Encoder, 0)
	filesPtr := make([]*os.File, 0)
	for _, value := range kva {
		hash := ihash(value.Key) % NReduce
		bucket := strconv.Itoa(hash)
		filesName := "m-temp-" + strconv.Itoa(workid) + "-" + bucket
		if encoder, ok := cacheFiles[bucket]; ok {
			_ = encoder.Encode(&value)
		} else {
			os.Remove(filesName)
			ioPtr, _ := os.OpenFile(filesName, os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0666)
			//转为json存入中间文件
			cacheFiles[bucket] = json.NewEncoder(ioPtr)
			_ = cacheFiles[bucket].Encode(&value)
			filesPtr = append(filesPtr, ioPtr)
		}
	}

	// 关闭文件指针
	for key, _ := range filesPtr {
		filesPtr[key].Close()
	}

	filesName := make([]string, 0)

	// 写入后重命名
	for bucket, _ := range cacheFiles {
		//filePtr.Close()
		filesName = append(filesName, "m-temp-"+strconv.Itoa(workid)+"-"+bucket)
	}

	mapDoneArgs := MapDoneArgs{}
	mapDoneReply := MapDoneReply{}

	mapDoneArgs.Workid = workid
	mapDoneArgs.Files = strings.Join(filesName, "_")
	call("Master.MapDone", &mapDoneArgs, &mapDoneReply)
	if mapDoneReply.Status == NEXT {
		return NEXT
	}
	return SUCCESS
}

func (w *worker) doReduce() int {
	// 以上，Map过程完成接下来进行 Reduce阶段
	// worker向 master请求 Reduce任务，如果没有请求到，则sleep，然后循环请求
	reduceArgs := ReduceArgs{}
	reduceReply := ReduceReply{}
	for {
		call("Master.GetReduce", &reduceArgs, &reduceReply)
		if reduceReply.Status == ERROR {
			return ERROR
		}
		if reduceReply.Start == true {
			break
		}
		time.Sleep(time.Second)
	}
	// 开始 reduce任务
	files := strings.Split(reduceReply.Files, "_")
	reduceId := reduceReply.ReduceId
	datas := make(map[string][]string, 0)
	sortDatas := make([]string, 0)
	for i := 0; i < len(files); i++ {
		filename := files[i]
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("in reduce cannot open %v", filename)
		}
		dec := json.NewDecoder(file)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err != nil {
				break
			}
			if _, ok := datas[kv.Key]; ok {
				datas[kv.Key] = append(datas[kv.Key], kv.Value)
			} else {
				datas[kv.Key] = []string{kv.Value}
				sortDatas = append(sortDatas, kv.Key)
			}
		}
		_ = file.Close()
	}

	os.Remove("mr-out-temp-" + strconv.Itoa(reduceId))
	ioPtr, _ := os.Create("mr-out-temp-" + strconv.Itoa(reduceId))

	for i := 0; i < len(sortDatas); i++ {
		key := sortDatas[i]
		values := datas[key]
		fmt.Fprintf(ioPtr, "%v %v\n", key, w.reducef(key, values))
	}
	ioPtr.Close()

	//os.Rename("mr-out-temp-"+strconv.Itoa(reduceId), "mr-out-"+strconv.Itoa(reduceId))

	doneArgs := ReduceDoneArgs{}
	doneReply := ReduceDoneReply{}
	doneArgs.Workid = reduceId
	doneArgs.File = "mr-out-temp-" + strconv.Itoa(reduceId)
	call("Master.ReduceDone", &doneArgs, &doneReply)
	return doneReply.Status
}

//
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//
//func CallExample() {
//
//	// declare an argument structure.
//	args := ExampleArgs{}
//
//	// fill in the argument(s).
//	args.X = 99
//
//	// declare a reply structure.
//	reply := ExampleReply{}
//
//	// send the RPC request, wait for the reply.
//	call("Master.TextRPC", &args, &reply)
//
//	// reply.Y should be 100.
//}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	//c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		//log.Fatal("dialing:", err)
		os.Exit(1)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	return false
}
